/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.channel.ram;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.sabayframework.channel.Lock;
import org.sabayframework.channel.impl.AbstractChannel;

/**
 * Concurrent implementation of Channel.
 * No limit on size of channel or lock, limited only by available RAM.
 * @author Alex Antonau
 * */
public class LinkedListChannel<T> extends AbstractChannel<T> {

	private static class Node{
		Object item;
		Node next;
	}
	private Node first;
	private Node last;
    /** Number of items in the queue */
    private volatile int count;
    /** closed flag */
    private volatile boolean closed;

    /** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    private final Condition notFull;

    /**
     * Creates channel from given array of elements.
     * @SafeVarargs 
     * */
    public LinkedListChannel(T... items) {
		this(false, items);
	}
    /**
     * Creates channel from given array of elements.
     * @SafeVarargs 
     * @param isBlocking true if this channel should support blocking put operation
     * @param items array of items
     * */
    public LinkedListChannel(boolean isBlocking, T... items) {
		super();
		if(items!=null){
			for(int i=0;i<items.length;++i)	addItem(items[i]);
			count = items.length;
		}
        this.lock = new ReentrantLock();
        notEmpty = lock.newCondition();
        notFull = isBlocking ? lock.newCondition() : null;
	}

    private void addItem(T item){
    	if(first == null){
    		first = new Node();
    		first.item = item;
    		last = first;
    	}else{
			last.next = new Node();
			last = last.next;
			last.item = item;
    	}
    }
    /**
     * Creates an empty channel 
     * */
	public LinkedListChannel(boolean isBlocking) {
		this(isBlocking,(T[]) null);
	}
	
    /**
     * Creates an empty channel
     * */
	public LinkedListChannel() {
		this(false);
	}

	private final class ReadLock implements Lock<T>{
    	private int locked;
		ReadLock(final int size) {
			count-=size;
			locked = size;
		}
		@Override
		public int size() {
			return locked;
		}

		@SuppressWarnings("unchecked")
		@Override
		public T get(int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			return (T) n.item;
		}
		@Override
		public void set(int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			n.item = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			Node n = first;
			for(int i=0;i<locked;++i){
				res[i] = n.item;
				n = n.next;
			}
			return res;
		}
		@Override
		public void unlock() {
			if(locked!=-1){
				if(count == 0) first = last = null;
				else{
					Node n = first;
					for(int i=0;i<locked;++i) n = n.next;
					first = n;
				}
				locked = -1;
				if(notFull != null) notFull.signalAll();
				lock.unlock();
			}
		}
		@Override
		public void cancel() {
			if(locked != -1){
				count+=locked;
				locked = -1;
				lock.unlock();
			}
		}
    }
    
	@Override
	public Lock<T> getReadLock(final int size) {
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		ReadLock result = null;
		if(lock.tryLock()){
			try{
				if(size <= count){
					result = new ReadLock(size);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}

	private final class WriteLock implements Lock<T>{
		private int locked;
		private Node first;
		
		WriteLock(int size) {
			count+=size;
			locked = size;
			Node oldLast = last;
			while(size-->0)addItem(null);
			first = (oldLast == null)? 
					LinkedListChannel.this.first:
					oldLast.next;
		}
		@Override
		public int size() {
			return locked;
		}
		@SuppressWarnings("unchecked")
		@Override
		public T get(int index) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			return (T) n.item;
		}
		@Override
		public void set(int index,final T elem) {
			if(index < 0 || index>=locked) throw new IndexOutOfBoundsException(String.valueOf(index));
			Node n = first;
			while(index-->0) n = n.next;
			n.item = elem;
		}
		@Override
		public Object[] toArray() {
			Object[] res = new Object[locked];
			Node n = first;
			for(int i=0;i<locked;++i){
				res[i] = n.item;
				n = n.next;
			}
			return res;
		}

		@Override
		public void unlock() {
			if(locked!=-1){
				locked = -1;
				first = null;
				notEmpty.signalAll();
				lock.unlock();
			}
		}
		@Override
		public void cancel() {
			if(locked!=-1){
				count-=locked;
				if(count ==0){
					LinkedListChannel.this.first = null;
					LinkedListChannel.this.last = null;
				}else if(locked > 0){
					Node n = LinkedListChannel.this.first;
					while(n.next != first) n=n.next;
					if(n!=null) {
						n.next = null;
						LinkedListChannel.this.last = n;
					}
				}
				locked = -1;
				lock.unlock();
			}
		}
	}
	@Override
	public Lock<T> getWriteLock(final int size) {
		if(size < 0) throw new IllegalArgumentException(String.format("Negative lock size %d", size));
		WriteLock result = null;
		if(lock.tryLock()){
			try{
				if(size <= freeSize()){
					result = new WriteLock(size);
				}
			} finally {
				if(result == null) lock.unlock();
			}
		}
		return result;
	}

	@Override
	public boolean isEmpty() {
		return count == 0;
	}

	@Override
	public boolean isFull() {
		return count >= capacity();
	}

	@Override
	public int size() {
		return count;
	}

	@Override
	public int capacity() {
		return Integer.MAX_VALUE;
	}

	@Override
	public int freeSize() {
		if(closed)return 0;
		return capacity()-count;
	}

	@Override
	public void close() {
		if(closed) return;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
       		closed = true;
       		notEmpty.signalAll();
       		if(notFull!=null) notFull.signalAll();
        } finally {
            lock.unlock();
        }
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

	@Override
	public Lock<T> waitForReadLock(int size, long timeout)	throws InterruptedException {
        final ReentrantLock lock = this.lock;
        long time=System.currentTimeMillis();
        ReadLock readLock = null;
        if(lock.tryLock(timeout, TimeUnit.MILLISECONDS)){
	        try {
            	timeout -= (System.currentTimeMillis()-time);
	            try {
	                while (timeout > 0 && !closed && count < size){
	                	time=System.currentTimeMillis();
	                	notEmpty.await(timeout, TimeUnit.MILLISECONDS);
	                	timeout -= (System.currentTimeMillis()-time);
	                }
	            } catch (InterruptedException ie) {
	            	notEmpty.signal(); // propagate to non-interrupted thread
	                throw ie;
	            }
				if(size <= count && !closed){
					readLock = new ReadLock(size);
					return readLock;
				}
	            
	        } finally {
	            if(readLock == null) lock.unlock();
	        }
        }
        return getReadLock(size);
	}

	/**
	 * Returns elements are currently in the channel.
	 * Not a thread safe method.
	 * */
	@SuppressWarnings("unchecked")
	public T[] toArray() {
		T[] res = (T[])new Object[count];
		Node n = first;
		for(int i=0;i<res.length;++i){
			res[i] = (T) n.item;
			n = n.next;
		}
		return res;
	}

	/**
	 * Puts given item into the channel. Waits while space is available.
	 * @throws IllegalStateException if channel is closed or interrupted
	 * */
	public void put(T elem){
		if(notFull == null) throw new IllegalStateException("Channel was created without blocking call support");
		lock.lock();
		try{
            try {
                while (!closed && Integer.MAX_VALUE == count){
                	notFull.await();
                }
            } catch (InterruptedException ie) {}
			
			if(freeSize() >= 1){
				addItem(elem);
				count++;
				notEmpty.signalAll();
			}else{
				throw new IllegalStateException(); 
			}
			
		}finally{
			lock.unlock();
		}
	}
}
